package com.atguigu.app.func;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.bean.TableProcess;
import com.atguigu.common.Constant;
import com.atguigu.util.HBaseUtil;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.hadoop.hbase.client.Connection;

public class DimCreateTableMapFunction extends RichMapFunction<String, TableProcess> {

    private Connection connection;

    @Override
    public void open(Configuration parameters) throws Exception {
        connection = HBaseUtil.getConnection();
    }

    //value:{"before":null,"after":{"source_table":"base_category3","sink_table":"dim_base_category3","sink_columns":"id,name,category2_id","sink_pk":"id","sink_extend":null},"source":{"version":"1.5.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1669162876406,"snapshot":"false","db":"gmall-220623-config","sequence":null,"table":"table_process","server_id":0,"gtid":null,"file":"","pos":0,"row":0,"thread":null,"query":null},"op":"r","ts_ms":1669162876406,"transaction":null}
    @Override
    public TableProcess map(String value) throws Exception {

        JSONObject jsonObject = JSON.parseObject(value);
        //获取op字段
        String op = jsonObject.getString("op");
        TableProcess tableProcess;
        if ("d".equals(op)) {
            tableProcess = JSON.parseObject(jsonObject.getString("before"), TableProcess.class);
        } else {
            tableProcess = JSON.parseObject(jsonObject.getString("after"), TableProcess.class);
        }
        tableProcess.setOp(op);

        if ("dim".equals(tableProcess.getSinkType())) {
            //建表 删表
            if ("d".equals(op)) {
                HBaseUtil.dropTable(connection, Constant.HBASE_NAME_SPACE, tableProcess.getSinkTable());
            } else if ("u".equals(op)) {
                HBaseUtil.dropTable(connection, Constant.HBASE_NAME_SPACE, tableProcess.getSinkTable());
                byte[][] splitKeys = HBaseUtil.getSplitKeys(tableProcess.getSinkExtend());
                HBaseUtil.createTable(connection, Constant.HBASE_NAME_SPACE, tableProcess.getSinkTable(), splitKeys, tableProcess.getSinkFamily());
            } else {
                byte[][] splitKeys = HBaseUtil.getSplitKeys(tableProcess.getSinkExtend());
                HBaseUtil.createTable(connection, Constant.HBASE_NAME_SPACE, tableProcess.getSinkTable(), splitKeys, tableProcess.getSinkFamily());
            }
        }

        return tableProcess;
    }

    @Override
    public void close() throws Exception {
        connection.close();
    }
}
